package yuekao5.dws;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import yuekao5.entity.Tm6;
import yuekao5.util.AsyncIOUtil;
import yuekao5.util.ClickHouseUtil;
import yuekao5.util.KafkaUtil;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class WideTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //DWS层汇总数据）编写Flink 流式程序，实时消费DWD层用户下单事实表（dwd-order-detail）数据，每日实时统计各个城市下单次数和金额（每隔1分钟实时计算一次），结果打印控制塔，考虑时区问题和窗口增量计算优化性能；（5分）
//        env.addSource(KafkaUtil.kafkaSource("dwd-order-detail")).print();
        SingleOutputStreamOperator<Tm6> orderinfodata = env.addSource(KafkaUtil.kafkaSources("dwd-order-detail")).map(new MapFunction<String, Tm6>() {
            @Override
            public Tm6 map(String s) throws Exception {
                String data = JSON.parseObject(s).getString("data");
                return JSON.parseObject(data, Tm6.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<Tm6>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> {
                    try {
                        return new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));
//        orderinfodata.print();

        OutputTag<Tm6> outputTag = new OutputTag<Tm6>("outputTag") {
        };

//        SingleOutputStreamOperator<Tuple2<String, String>> process = orderinfodata.keyBy(x -> x.getSender_city_id())
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .process(new ProcessWindowFunction<Tm6, Tuple2<String, String>, Integer, TimeWindow>() {
//                    @Override
//                    public void process(Integer integer, ProcessWindowFunction<Tm6, Tuple2<String, String>, Integer, TimeWindow>.Context context, Iterable<Tm6> iterable, Collector<Tuple2<String, String>> collector) throws Exception {
//                        String eet = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").format(context.window().getStart());
//                        String ent = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").format(context.window().getEnd());
//                        collector.collect(new Tuple2<>(eet, ent));
//                    }
//                });
//        process.print();


        SingleOutputStreamOperator<Tm6> process = orderinfodata.keyBy(new KeySelector<Tm6, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> getKey(Tm6 tm6) throws Exception {
                        return new Tuple2<>(tm6.getCreate_time(),tm6.getSender_city_id());
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(outputTag)
                        .process(new ProcessWindowFunction<Tm6, Tm6, Tuple2<String, Integer>, TimeWindow>() {
                            @Override
                            public void process(Tuple2<String, Integer> stringIntegerTuple2, ProcessWindowFunction<Tm6, Tm6, Tuple2<String, Integer>, TimeWindow>.Context context, Iterable<Tm6> iterable, Collector<Tm6> collector) throws Exception {
                                HashSet<Integer> set = new HashSet<>();
                                Integer sender_city_id = 0;
                                String name = null;
                                Double amount = 0.0;
                                String create_time = null;
                                for (Tm6 o : iterable) {
                                    set.add(o.getId());
                                    sender_city_id = o.getSender_city_id();
                                    amount += amount;
                                    create_time = o.getCreate_time();
                                }
                                String eet = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").format(context.window().getStart());
                                String ent = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'").format(context.window().getEnd());
                                collector.collect(new Tm6(eet, ent, set.size(), sender_city_id, name, amount, create_time));
                            }
                        });


//        process.print();
//        process.getSideOutput(outputTag).print();
//        process.getSideOutput(outputTag).map(x->JSON.toJSONString(x)).addSink(KafkaUtil.kafkaSink("tms-late-topic"));

        //7）、（DWS层汇总数据）采用异步IO方式，关联HBase数据库中维表：dim_base_region_info，补充第6题计算结果中城市维度信息（获取城市名称），
        // 并在ClickHouse数据库中创建表（选择合适表引擎，实现upsert插入更新功能），将分析结果保存ClickHouse表：trade_order_province_day中，使用SELECT查询统计指标；（5分）
        SingleOutputStreamOperator<Tm6> tm6SingleOutputStreamOperator = AsyncDataStream.unorderedWait(process, new AsyncIOUtil(), 1000, TimeUnit.MILLISECONDS, 100);
//        tm6SingleOutputStreamOperator.print();
        tm6SingleOutputStreamOperator.addSink(new ClickHouseUtil());
        env.execute();
    }
}
